package com.banmao.rabbitmq.c4_confirm;

import com.alibaba.fastjson2.JSON;
import com.banmao.rabbitmq.utils.RabbitConstant;
import com.banmao.rabbitmq.utils.RabbitUtil;
import com.rabbitmq.client.*;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.LinkedHashMap;
import java.util.Map;

@Slf4j
public class WeatherBureau {

    public static void main(String[] args) throws Exception {
        LinkedHashMap<String, String> areas = new LinkedHashMap<>() {
            {
                put("china.guangdong.guangzhou.20991011", "中国广东广州20991011天气");
                put("china.hebei.shijiazhuang.20991011", "中国河北石家庄20991011天气");
                put("china.guangdong.guangzhou.20991012", "中国广东广州20991012天气");
                put("china.hebei.shijiazhuang.20991012", "中国河北石家庄20991012天气");
            }
        };

        Connection connection = RabbitUtil.getConnection();
        Channel channel = connection.createChannel();

        // 开启 Confirm 监听模式
        channel.confirmSelect();
        // 如果要监听 Confirm，就不能关闭 channel 资源
        channel.addConfirmListener(new ConfirmListener() {
            /**
             * Broker 返回 ack 时的回调，不关心是否投递到队列
             * @param deliveryTag 此消息ID
             * @param multiple 消息是否为批量接收（一般用不到
             */
            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                log.info("消息已被 Broker 接收, Tag: {}", deliveryTag);
            }

            /**
             * Broker 返回 nack 消息拒收时的回调
             * @param deliveryTag 此消息ID
             * @param multiple 消息是否为批量接收（一般用不到
             */
            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                log.info("消息已被 Broker 拒收, Tag: {}", deliveryTag);
            }
        });

        channel.addReturnListener(new ReturnListener() {
            @Override
            public void handleReturn(int replyCode, String replyText, String exchange, String routingKey,
                                     AMQP.BasicProperties properties, byte[] body) throws IOException {
                log.error("消息已被 Broker 退回, replyCode:{}, replyText:{}, exchange:{}, routingKey:{}, properties:{}, body:{}",
                        replyCode, replyText, exchange, routingKey, JSON.toJSONString(properties), new String(body));
            }
        });

        channel.addReturnListener(new ReturnCallback() {
            @Override
            public void handle(Return returnMessage) {
                log.error("消息已被 Broker 退回, returnMessage:{}", JSON.toJSONString(returnMessage));
            }
        });

        for (Map.Entry<String, String> entry : areas.entrySet()) {
            String key = entry.getKey();
            String value = entry.getValue();
            // 第三个参数-mandatory：true-代表刚消息无法正常投递则 return 回生产者，false-代表直接丢弃
            // 默认 false
            channel.basicPublish(RabbitConstant.EXCHANGE_WATHER_TOPIC, key, true, null, value.getBytes());
        }

//        channel.close();
//        connection.close();
    }
}
